package com.zhi.common.utils.thread;

import com.google.common.collect.Lists;
import com.zhi.common.exception.base.BaseException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * <p>
 * 异步并发处理工具类
 * </p>
 * @version CompletableFutureUtil.java v1.0 2021/6/30 下午3:42
 */
public class CompletableFutureUtil {

    /**
     * 分批聚合异步处理方法,结果为List类型
     * @param listParam 需要按照指定大小拆分批次的list
     * @param partitionSize 批次大小
     * @param executorService 线程池
     * @param requestFunc 请求方法
     * @param <T>
     * @param <R>
     * @return 结果为List类型
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R> List<R> partitionAsyncInvokeWithListResult(List<T> listParam,
                                                                    int partitionSize,
                                                                    Executor executorService,
                                                                    Function<List<T>, List<R>> requestFunc)
            throws ExecutionException, InterruptedException {
        if (CollectionUtils.isEmpty(listParam)) {
            return Collections.emptyList();
        }
        List<R> res = partitionAsyncInvoke(listParam, partitionSize, executorService, requestFunc, listMergeFunc());
        if (CollectionUtils.isEmpty(res)) {
            return Collections.emptyList();
        }
        return res;
    }

    public static <T, R> List<R> partitionAsyncInvokeWithListResultQuietly(List<T> listParam,
                                                                    int partitionSize,
                                                                    Executor executorService,
                                                                    Function<List<T>, List<R>> requestFunc) {
        try {
            return partitionAsyncInvokeWithListResult(listParam, partitionSize, executorService, requestFunc);
        } catch (ExecutionException | InterruptedException e) {
            if (e.getCause() instanceof BaseException) {
                throw (BaseException)e.getCause();
            }
            throw new BaseException("系统错误");
        }
    }

    /**
     * 分批聚合异步处理方法,结果为List类型
     * @param originRequest  原始请求
     * @param requestSplitter 请求拆分器
     * @param executorService 线程池
     * @param requestFunc 请求方法
     * @param <T>
     * @param <R>
     * @return 结果为List类型
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R> List<R> partitionAsyncInvokeWithListResult(T originRequest,
                                                                    Function<T, List<T>> requestSplitter,
                                                                    Executor executorService,
                                                                    Function<T, List<R>> requestFunc)
            throws ExecutionException, InterruptedException {
        if (originRequest == null) {
            return Collections.emptyList();
        }
        List<R> res = partitionAsyncInvoke(originRequest, requestSplitter, executorService, requestFunc, listMergeFunc());
        if (CollectionUtils.isEmpty(res)) {
            return Collections.emptyList();
        }
        return res;
    }

    public static <T, R> List<R> partitionAsyncInvokeWithListResultQuietly(T originRequest,
                                                                    Function<T, List<T>> requestSplitter,
                                                                    Executor executorService,
                                                                    Function<T, List<R>> requestFunc) {
        try {
            return partitionAsyncInvokeWithListResult(originRequest, requestSplitter, executorService, requestFunc);
        } catch (ExecutionException | InterruptedException e) {
            if (e.getCause() instanceof BaseException) {
                throw (BaseException)e.getCause();
            }
            throw new BaseException("系统错误");
        }
    }


    /**
     * 分批聚合异步处理方法,结果为简单Map类型
     * 注意事项: 如果key相同会直接覆盖!
     * @param listParam 需要按照指定大小拆分批次的list
     * @param partitionSize 批次大小
     * @param executorService 线程池
     * @param requestFunc 请求方法
     * @param <T>
     * @param <R>
     * @return 结果为List类型
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R, U> Map<R, U> partitionAsyncInvokeWithMapResult(List<T> listParam,
                                                                        int partitionSize,
                                                                        Executor executorService,
                                                                        Function<List<T>, Map<R, U>> requestFunc)
            throws ExecutionException, InterruptedException {
        if (CollectionUtils.isEmpty(listParam)) {
            return Collections.emptyMap();
        }
        Map<R, U> res = partitionAsyncInvoke(listParam, partitionSize, executorService, requestFunc, mapMergeFunc());
        if (MapUtils.isEmpty(res)) {
            return Collections.emptyMap();
        }
        return res;
    }

    public static <T, R, U> Map<R, U> partitionAsyncInvokeWithMapResultQuietly(List<T> listParam,
                                                                        int partitionSize,
                                                                        Executor executorService,
                                                                        Function<List<T>, Map<R, U>> requestFunc) {
        try {
            return partitionAsyncInvokeWithMapResult(listParam, partitionSize, executorService, requestFunc);
        } catch (ExecutionException | InterruptedException e) {
            if (e.getCause() instanceof BaseException) {
                throw (BaseException)e.getCause();
            }
            throw new BaseException("系统错误");
        }
    }

    /**
     * 分批聚合异步处理方法,结果为简单Map类型
     * 注意事项: 如果key相同会直接覆盖!
     * @param originRequest  原始请求
     * @param requestSplitter 请求拆分器
     * @param executorService 线程池
     * @param requestFunc 请求方法
     * @param <T>
     * @param <R>
     * @return 结果为List类型
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R, U> Map<R, U> partitionAsyncInvokeWithMapResult(T originRequest,
                                                                        Function<T, List<T>> requestSplitter,
                                                                        Executor executorService,
                                                                        Function<T, Map<R, U>> requestFunc)
            throws ExecutionException, InterruptedException {
        if (originRequest == null) {
            return Collections.emptyMap();
        }
        Map<R, U> res = partitionAsyncInvoke(originRequest, requestSplitter, executorService, requestFunc, mapMergeFunc());
        if (MapUtils.isEmpty(res)) {
            return Collections.emptyMap();
        }
        return res;
    }


    public static <T, R, U> Map<R, U> partitionAsyncInvokeWithMapResultQuietly(T originRequest,
                                                                        Function<T, List<T>> requestSplitter,
                                                                        Executor executorService,
                                                                        Function<T, Map<R, U>> requestFunc) {
        try {
            return partitionAsyncInvokeWithMapResult(originRequest, requestSplitter, executorService, requestFunc);
        } catch (ExecutionException | InterruptedException e) {
            if (e.getCause() instanceof BaseException) {
                throw (BaseException)e.getCause();
            }
            throw new BaseException("系统错误");
        }
    }

    /**
     * 分批聚合异步处理方法
     * @param listParam 需要分批处理的参数对象
     * @param partitionSize 每批大小
     * @param executorService 异步执行的线程池,必须传入
     * @param requestFunc 执行请求的函数
     * @param mergeFunc 多个结果的合并方法
     * @param <T>
     * @param <R>
     * @return merge结果可能为 null
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R> R partitionAsyncInvoke(List<T> listParam,
                                                int partitionSize,
                                                Executor executorService,
                                                Function<List<T>, R> requestFunc,
                                                BinaryOperator<R> mergeFunc)
            throws ExecutionException, InterruptedException {
        if (CollectionUtils.isEmpty(listParam)) {
            return null;
        }
        if (partitionSize < 0 || executorService == null || requestFunc == null || mergeFunc == null) {
            throw new BaseException("参数错误");
        }

        List<CompletableFuture<R>> completableFutures = Lists.partition(listParam, partitionSize)
                .stream()
                .map(eachList -> CompletableFuture.supplyAsync(() -> requestFunc.apply(eachList), executorService))
                .collect(Collectors.toList());

        CompletableFuture<Void> allFinished = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[]{}));
        allFinished.get();
        return completableFutures.stream()
                .map(CompletableFuture::join)
                .filter(Objects::nonNull)
                .reduce(mergeFunc)
                .orElse(null);
    }

    /**
     * 自定义拆分规则并发请求处理
     * @param originRequest  原始请求
     * @param requestSplitter 请求拆分器
     * @param executorService 线程池
     * @param requestFunc remote方法
     * @param mergeFunc 合并函数
     * @param <T>
     * @param <R>
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static <T, R> R partitionAsyncInvoke(T originRequest,
                                                Function<T, List<T>> requestSplitter,
                                                Executor executorService,
                                                Function<T, R> requestFunc,
                                                BinaryOperator<R> mergeFunc)
            throws ExecutionException, InterruptedException {
        if (originRequest == null) {
            return null;
        }
        if (requestSplitter == null || executorService == null || requestFunc == null || mergeFunc == null) {
            throw new BaseException("参数错误");
        }
        List<T> subReqList = requestSplitter.apply(originRequest);
        List<CompletableFuture<R>> completableFutures = subReqList
                .stream()
                .map(eachList -> CompletableFuture.supplyAsync(() -> requestFunc.apply(eachList), executorService))
                .collect(Collectors.toList());

        CompletableFuture<Void> allFinished = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[]{}));
        allFinished.get();
        return completableFutures.stream()
                .map(CompletableFuture::join)
                .filter(Objects::nonNull)
                .reduce(mergeFunc)
                .orElse(null);
    }

    /**
     * list合并函数
     * @param <R>
     * @return
     */
    private static <R> BinaryOperator<List<R>> listMergeFunc() {
        return (l, r) -> {
            if (CollectionUtils.isNotEmpty(l) && CollectionUtils.isNotEmpty(r)) {
                CollectionUtils.addAll(l, r);
            }
            if (CollectionUtils.isNotEmpty(l)) {
                return l;
            }
            return r;
        };
    }

    /**
     * map合并函数
     * @param <R>
     * @param <U>
     * @return
     */
    private static <R, U> BinaryOperator<Map<R, U>> mapMergeFunc() {
        return (l, r) -> {
            if (MapUtils.isNotEmpty(l) && MapUtils.isNotEmpty(r)) {
                l.putAll(r);
            }
            if (MapUtils.isNotEmpty(l)) {
                return l;
            }
            return r;
        };
    }

}
